package com.shaozj.demo.kafka.consumer;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import java.util.Objects;


@Slf4j
@RestController
public class ConsumerController {

//    private final KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

//    public ConsumerController(KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry) {
//        this.kafkaListenerEndpointRegistry = kafkaListenerEndpointRegistry;
//    }

//    @KafkaListener(topics = {"topic1", "topic2"}, errorHandler = "myKafkaListenerErrorHandler")
//    public void listen1(ConsumerRecord<Object, Object> consumerRecord, Acknowledgment ack) {
//        try {
//            //用于异常处理
//            log.info("", consumerRecord.value());
//            //手动确认
//            ack.acknowledge();
//        } catch(Exception e) {
//            log.error("消费失败：{}", e);
//        }
//    }
//
//    /**
//     * 下面的方法可以手动操控 kafka 的队列监听情况
//     * 先发一条消息，因为 autoStartup = false，所以并不会看到有消息进入监听器
//     * 接着启动监听器，。start/webGroup，可以看到一条消息你来了，
//     * pause 暂停监听，resume 是继续监听
//     * @param listenerId
//     */
//    @GetMapping("/pause/{listenerId}")
//    public void stop(@PathVariable  String listenerId) {
//        Objects.requireNonNull(kafkaListenerEndpointRegistry.getListenerContainer(listenerId)).pause();
//    }
//
//    @GetMapping("/resume/{listenerId}")
//    public void resume(String listenerId) {
//        Objects.requireNonNull(kafkaListenerEndpointRegistry.getListenerContainer(listenerId)).resume();
//    }
//
//    @GetMapping("/start/{listenerId}")
//    public void start(@PathVariable String listenerId) {
//        Objects.requireNonNull(kafkaListenerEndpointRegistry.getListenerContainer(listenerId)).start();
//
//    }

}
